#include "sys.h"
#include "log.h"
#include "thread_pool.h"
#include "dial.h"
#include "cJSON.h"
#include "httpvideodown_tactics.h"
#include "data_report.h"
#include "trans.h"

#define RESET_MIN_MAX(min, max) \
    (min = 0);                  \
    (max = 0)

static float Max_rate = 0, Min_rate = 0;
static int* Valid_bytes_ptr;
static int Valid_bytes_count;
static int Ignore_time, Total_time, Interval_time;

struct down_up_urls {
    int up_count, down_count;
    char** up;
    char** down;
};

static int _get_cur_bytes()
{
    int i = 0;
    int bytes = 0;
    for (i = 0; i < Valid_bytes_count; i++) {
        bytes += Valid_bytes_ptr[i];
    }
    return bytes;
}
static void* _cal_max_min_rate(void* arg)
{
    int i = 0;
    float cur_rate = 0;
    int startTime = 0;
    int prev_bytes, cur_bytes;
    int total_time = Total_time;

    usleep(MSEC2USEC(Ignore_time));
    prev_bytes = _get_cur_bytes();
    total_time -= Ignore_time;
    while ((total_time -= Interval_time) > 0) {
        usleep(MSEC2USEC(Interval_time));
        cur_bytes = _get_cur_bytes();
        cur_rate = (float)(cur_bytes - prev_bytes) * 8 / MSEC2SEC(Interval_time);
        Max_rate = MAX_VAL(cur_rate, Max_rate);
        Min_rate = Min_rate ? MIN_VAL(cur_rate, Min_rate) : cur_rate;
        prev_bytes = cur_bytes;
    }
}
static void _free_down_up_urls(struct down_up_urls* urls)
{
    int i = 0;

    for (i = 0; i < urls->up_count; i++) {
        free(urls->up[i]);
    }
    for (i = 0; i < urls->down_count; i++) {
        free(urls->down[i]);
    }
    if (urls->down)
        free(urls->down);
    if (urls->up)
        free(urls->up);
}
static int _get_url(char* url, struct down_up_urls* urls)
{
    int ret = 1, count = 0;
    char *delim = ",", *tmp_url, *saveptr;
    char* ptr;

    if (strlen(url) == 0)
        return 0;

    tmp_url = strdup(url);
    urls->down = calloc(32, sizeof(char*));
    ptr = strtok_r(tmp_url, delim, &saveptr);
    if (ptr) {
        urls->down[count++] = strdup(ptr);
    }
    while ((ptr = strtok_r(NULL, delim, &saveptr)) != NULL) {
        urls->down[count++] = strdup(ptr);
    }
    urls->down_count = count;

    free(tmp_url);
    return ret;
}

void* do_httpvideodown_tactics(void* arg, char* nodeUrl, char* probeId)
{

    httpvideodown_tactics_t* httpvideodown = (httpvideodown_tactics_t*)arg;
    httpvideodownTactics_t* ptr;
    struct httpvideodownTactics_head* head = &httpvideodown->head;
    cJSON *result, *obj, *arr;
    struct down_up_urls urls;

    dial_arg_t dial_arg = { 0 };
    dial_result_t* dial_result = NULL;
    http_result_t* httpdown_result = NULL;
    char tmp_str[32] = { 0 };
    char topic[32] = { 0 };
    char *post_data = NULL, reportTime[32], collectTime[32];
    int i = 0;
    int down_success = 0;

    if (httpvideodown->count == 0) {
        return NULL;
    }
    memset(&urls, 0, sizeof(urls));

    result = cJSON_CreateObject();
    cJSON_AddStringToObject(result, "dialTaskId", httpvideodown->taskId);
    cJSON_AddStringToObject(result, "version", httpvideodown->version);
    cJSON_AddStringToObject(result, "indexType", "httpvideodown");
    get_timestring(reportTime);
    cJSON_AddStringToObject(result, "reportTime", reportTime);
    get_timestring(collectTime);
    cJSON_AddStringToObject(result, "collectTime", collectTime);
    // pthread_mutex_lock(&http->mutex);
    arr = cJSON_CreateArray();


    TAILQ_FOREACH(ptr, head, entry)
    {
        if (!_get_url(ptr->urls, &urls)) {
            continue;
        }
        for (i = 0; i < urls.down_count; i++) {

            SPEEDTEST_LOCK();

            obj = cJSON_CreateObject();
            
            memset(&dial_arg, 0, sizeof(dial_arg));

            RESET_MIN_MAX(Min_rate, Max_rate);
            // dial_arg.in.http_arg.uris = urls.down;
            // dial_arg.in.http_arg.uris_count = urls.down_count;
            // dial_arg.in.http_arg.threads = ptr->threads;
            char* tmp_url[1];
            tmp_url[0] = strdup(urls.down[i]);
            dial_arg.in.http_arg.uris = tmp_url;
            dial_arg.in.http_arg.uris_count = 1;
            dial_arg.in.http_arg.threads = ptr->threads;
            Valid_bytes_count = dial_arg.in.http_arg.threads;
            Valid_bytes_ptr = calloc(Valid_bytes_count, sizeof(int));
            dial_arg.in.http_arg.valid_bytes_ptr = Valid_bytes_ptr;
            Total_time = dial_arg.in.http_arg.duration = SEC2MSEC(ptr->downTime);
            Interval_time = dial_arg.in.http_arg.interval = SEC2MSEC(1);
            Ignore_time = dial_arg.in.http_arg.ignore_duration = SEC2MSEC(ptr->ignoreTime);

            pthread_t thid;
            pthread_create(&thid, NULL, _cal_max_min_rate, NULL);
            dial_result = (dial_result_t*)dial_http_download(&dial_arg);
            dial_result->out.http_result.min_rate = Min_rate;
            dial_result->out.http_result.max_rate = Max_rate;
            http_result_print(&dial_result->out.http_result);
    
            pthread_join(thid, NULL);

            cJSON_AddStringToObject(obj, "address", tmp_url[0]);
            sprintf(tmp_str, "%.0f", dial_result->out.http_result.latency);
            cJSON_AddStringToObject(obj, "firstPageTime", tmp_str);
            sprintf(tmp_str, "%.0f", KBps(dial_result->out.http_result.min_rate / 8));
            cJSON_AddStringToObject(obj, "downMinSpeed", tmp_str);
            sprintf(tmp_str, "%.0f", KBps(dial_result->out.http_result.max_rate / 8));
            cJSON_AddStringToObject(obj, "downMaxSpeed", tmp_str);
            sprintf(tmp_str, "%.0f", KBps(dial_result->out.http_result.rate / 8));
            cJSON_AddStringToObject(obj, "downAvgSpeed", tmp_str);
            cJSON_AddNumberToObject(obj, "downActualDuration", dial_result->out.http_result.total_time);
            cJSON_AddNumberToObject(obj, "downActualSize", dial_result->out.http_result.total_bytes);
        
            if(dial_result->out.http_result.rate > 0){
                down_success = 1;
            }

            log_info("Http Video :%s, err:%d", tmp_url[0], dial_result->err);

            free(dial_result);
            free(Valid_bytes_ptr);
            free(tmp_url[0]);
    
            strcpy(topic, ptr->kafkaTopic);
            cJSON_AddStringToObject(obj, "businessTag", ptr->businessTag);
            get_timestring(collectTime);
            cJSON_AddStringToObject(obj, "collectTime", collectTime);
            
            if(down_success){
                cJSON_AddItemToArray(arr, obj);
            }else{
                cJSON_Delete(obj);
            }
            
            SPEEDTEST_UNLOCK();
        }
        _free_down_up_urls(&urls);
    }

    cJSON_AddItemToObject(result, "httpVideoSpeedList", arr);
    post_data = cJSON_Print(result);
    if(Debug_mode()){
        log_debug("httpvideodown post_data:%s", post_data);
    }
    // data_report(httpvideodown->kafka_url, topic, post_data, strlen(post_data) );
    free(post_data);
    // cJSON_Delete(result);
    // pthread_cond_signal(&httpvideodown->cond);
    // pthread_mutex_unlock(&httpvideodown->mutex);
    return (void*)result;
}